// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//


#include <pollux/connectors/hive/storage_adapters/s3fs/S3Config.h>
#include <pollux/common/config/config.h>

#include <gtest/gtest.h>

namespace kumo::pollux::filesystems {
namespace {
TEST(S3ConfigTest, defaultConfig) {
  auto config = std::make_shared<config::ConfigBase>(
      melon::F14FastMap<std::string, std::string>());
  auto s3Config = S3Config("", config);
  ASSERT_EQ(s3Config.useVirtualAddressing(), true);
  ASSERT_EQ(s3Config.useSSL(), true);
  ASSERT_EQ(s3Config.useInstanceCredentials(), false);
  ASSERT_EQ(s3Config.endpoint(), std::nullopt);
  ASSERT_EQ(s3Config.endpointRegion(), std::nullopt);
  ASSERT_EQ(s3Config.accessKey(), std::nullopt);
  ASSERT_EQ(s3Config.secretKey(), std::nullopt);
  ASSERT_EQ(s3Config.iamRole(), std::nullopt);
  ASSERT_EQ(s3Config.iamRoleSessionName(), "pollux-session");
  ASSERT_EQ(s3Config.payloadSigningPolicy(), "Never");
  ASSERT_EQ(s3Config.cacheKey("foo", config), "foo");
  ASSERT_EQ(s3Config.bucket(), "");
}

TEST(S3ConfigTest, overrideConfig) {
  melon::F14FastMap<std::string, std::string> configFromFile = {
      {S3Config::baseConfigKey(S3Config::Keys::kPathStyleAccess), "true"},
      {S3Config::baseConfigKey(S3Config::Keys::kSSLEnabled), "false"},
      {S3Config::baseConfigKey(S3Config::Keys::kUseInstanceCredentials),
       "true"},
      {"hive.s3.payload-signing-policy", "RequestDependent"},
      {S3Config::baseConfigKey(S3Config::Keys::kEndpoint), "endpoint"},
      {S3Config::baseConfigKey(S3Config::Keys::kEndpointRegion), "region"},
      {S3Config::baseConfigKey(S3Config::Keys::kAccessKey), "access"},
      {S3Config::baseConfigKey(S3Config::Keys::kSecretKey), "secret"},
      {S3Config::baseConfigKey(S3Config::Keys::kIamRole), "iam"},
      {S3Config::baseConfigKey(S3Config::Keys::kIamRoleSessionName), "pollux"},
      {S3Config::baseConfigKey(S3Config::Keys::kCredentialsProvider),
       "my-credentials-provider"}};
  auto configBase =
      std::make_shared<config::ConfigBase>(std::move(configFromFile));
  auto s3Config = S3Config("bucket", configBase);
  ASSERT_EQ(s3Config.useVirtualAddressing(), false);
  ASSERT_EQ(s3Config.useSSL(), false);
  ASSERT_EQ(s3Config.useInstanceCredentials(), true);
  ASSERT_EQ(s3Config.endpoint(), "endpoint");
  ASSERT_EQ(s3Config.endpointRegion(), "region");
  ASSERT_EQ(s3Config.accessKey(), std::optional("access"));
  ASSERT_EQ(s3Config.secretKey(), std::optional("secret"));
  ASSERT_EQ(s3Config.iamRole(), std::optional("iam"));
  ASSERT_EQ(s3Config.iamRoleSessionName(), "pollux");
  ASSERT_EQ(s3Config.payloadSigningPolicy(), "RequestDependent");
  ASSERT_EQ(s3Config.cacheKey("foo", configBase), "endpoint-foo");
  ASSERT_EQ(s3Config.cacheKey("bar", configBase), "endpoint-bar");
  ASSERT_EQ(s3Config.bucket(), "bucket");
  ASSERT_EQ(s3Config.credentialsProvider(), "my-credentials-provider");
}

TEST(S3ConfigTest, overrideBucketConfig) {
  std::string_view bucket = "bucket";
  melon::F14FastMap<std::string, std::string> bucketConfigFromFile = {
      {S3Config::baseConfigKey(S3Config::Keys::kPathStyleAccess), "true"},
      {S3Config::baseConfigKey(S3Config::Keys::kSSLEnabled), "false"},
      {S3Config::baseConfigKey(S3Config::Keys::kUseInstanceCredentials),
       "true"},
      {S3Config::baseConfigKey(S3Config::Keys::kEndpoint), "endpoint"},
      {S3Config::bucketConfigKey(S3Config::Keys::kEndpoint, bucket),
       "bucket.s3-region.amazonaws.com"},
      {S3Config::baseConfigKey(S3Config::Keys::kAccessKey), "access"},
      {S3Config::bucketConfigKey(S3Config::Keys::kAccessKey, bucket),
       "bucket-access"},
      {"hive.s3.payload-signing-policy", "Always"},
      {S3Config::baseConfigKey(S3Config::Keys::kSecretKey), "secret"},
      {S3Config::bucketConfigKey(S3Config::Keys::kSecretKey, bucket),
       "bucket-secret"},
      {S3Config::baseConfigKey(S3Config::Keys::kIamRole), "iam"},
      {S3Config::baseConfigKey(S3Config::Keys::kIamRoleSessionName), "pollux"},
      {S3Config::baseConfigKey(S3Config::Keys::kCredentialsProvider),
       "my-credentials-provider"},
      {S3Config::bucketConfigKey(S3Config::Keys::kCredentialsProvider, bucket),
       "override-credentials-provider"}};
  auto configBase =
      std::make_shared<config::ConfigBase>(std::move(bucketConfigFromFile));
  auto s3Config = S3Config(bucket, configBase);
  ASSERT_EQ(s3Config.useVirtualAddressing(), false);
  ASSERT_EQ(s3Config.useSSL(), false);
  ASSERT_EQ(s3Config.useInstanceCredentials(), true);
  ASSERT_EQ(s3Config.endpoint(), "bucket.s3-region.amazonaws.com");
  // Inferred from the endpoint.
  ASSERT_EQ(s3Config.endpointRegion(), "region");
  ASSERT_EQ(s3Config.accessKey(), std::optional("bucket-access"));
  ASSERT_EQ(s3Config.secretKey(), std::optional("bucket-secret"));
  ASSERT_EQ(s3Config.iamRole(), std::optional("iam"));
  ASSERT_EQ(s3Config.iamRoleSessionName(), "pollux");
  ASSERT_EQ(s3Config.payloadSigningPolicy(), "Always");
  ASSERT_EQ(
      s3Config.cacheKey(bucket, configBase),
      "bucket.s3-region.amazonaws.com-bucket");
  ASSERT_EQ(s3Config.cacheKey("foo", configBase), "endpoint-foo");
  ASSERT_EQ(s3Config.credentialsProvider(), "override-credentials-provider");
}

} // namespace
} // namespace kumo::pollux::filesystems
